package com.shiku.push;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.shiku.xmpppush.autoconfigure.ApplicationProperties.IMConfig;
import com.shiku.xmpppush.autoconfigure.ApplicationProperties.MQConfig;

/**
 * 
 * @Description: TODO(xmpp消费者)
 * @author Administrator
 * @date 2018年12月17日 下午3:09:21
 * @version V1.0
 */
//@Component
public class MessageMQConsumer implements InitializingBean{
	private static final Logger log = LoggerFactory.getLogger(MessageMQConsumer.class);
	
	private DefaultMQPushConsumer chatConsumer;

	@Autowired(required=false) 
	public MQConfig mqConfig;
	@Autowired(required=false) 
	public IMConfig imConfig;
	
	private List<String> sysUserList=null;
	
	private synchronized List<String> getUserList(){
		if(null!=sysUserList)
			return sysUserList;
		sysUserList=Collections.synchronizedList(new ArrayList<String>());
		
		return sysUserList;
	}
	
	@Override
	public void afterPropertiesSet() throws Exception {
		
		initConsumer();
	}
	
	public  void initConsumer() throws InterruptedException{
		log.info("Xmpp MQ config nameAddr ===> "+mqConfig.getNameAddr());
		DefaultMQPushConsumer consumer = getChatConsumer();
        /*consumer.registerMessageListener(
        		new SkMessageListenerConcurrently(mqConfig.getNameAddr(),imConfig));*/
        
        try {
	    	   consumer.subscribe("xmppMessage", "*");
	    	   consumer.start();
			} catch (Exception e) {
				e.printStackTrace();
			}
	}
	
	public DefaultMQPushConsumer getChatConsumer() {
		if(null!=chatConsumer)
			return chatConsumer;
		
		try {
			chatConsumer=new DefaultMQPushConsumer("xmppProducer");
			chatConsumer.setNamesrvAddr(mqConfig.getNameAddr());
			chatConsumer.setVipChannelEnabled(false);
			chatConsumer.setConsumeThreadMin(mqConfig.getThreadMin());
			chatConsumer.setConsumeThreadMax(mqConfig.getThreadMax());
			chatConsumer.setConsumeMessageBatchMaxSize(mqConfig.getBatchMaxSize());
			chatConsumer.setMessageModel(MessageModel.CLUSTERING);
		} catch (Exception e) {
			e.printStackTrace();
		}
		return chatConsumer;
	}
	
}
